package com.alibaba.nacos.common.notify;

import com.alibaba.nacos.common.notify.listener.MultiSubscriber;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class NotifyCenter {
    private static final NotifyCenter INSTANCE = new NotifyCenter();
    public static int ringBufferSize;
    private final Map<String, EventPublisher> publisherMap = new ConcurrentHashMap<>(16);

    static {
        ringBufferSize = 16384;
    }

    public static boolean publishEvent(Event event) {
        return publishEvent(event.getClass(), event);
    }

    private static boolean publishEvent(Class<? extends Event> eventType,
                                        Event event) {
        String key = eventType.getCanonicalName();
        EventPublisher publisher = INSTANCE.publisherMap.get(key);
        if (publisher != null) {
//            log.info("publish event:{} publisher:{}", eventType.getSimpleName(), publisher.getClass().getSimpleName());
            return publisher.publish(event);
        }

        log.info("publish event:{} publisher not found", eventType.getSimpleName());
        return false;
    }

    public static void registerSubscriber(Subscriber consumer,
                                          EventPublisherFactory factory) {
        if (consumer instanceof MultiSubscriber multiSubscriber) {
            for (Class<? extends Event> subscribeType : multiSubscriber.subscribeTypes()) {
                addSubscriber(consumer, subscribeType, factory);
            }
        }
    }

    private static void addSubscriber(Subscriber consumer,
                                      Class<? extends Event> subscribeType,
                                      EventPublisherFactory factory) {
        String topic = subscribeType.getCanonicalName();
        EventPublisher publisher;
        synchronized (NotifyCenter.class) {
            publisher = INSTANCE.publisherMap.computeIfAbsent(topic, k -> factory.apply(subscribeType, ringBufferSize));
        }

        if (publisher instanceof ShardedEventPublisher shardedEventPublisher) {
            shardedEventPublisher.addSubscriber(consumer, subscribeType);
        } else {
            publisher.addSubscriber(consumer);
        }
    }
}
